package com.ndp.fb.rdb.mongo.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.*;
import com.ndp.fb.constant.MongoConstants;
import com.ndp.fb.model.business.TimeRange;
import com.ndp.fb.model.report.param.EcReportForABTestInputParam;
import com.ndp.fb.model.report.param.EcReportInputFilterParam;
import com.ndp.fb.model.report.param.EcReportInputParam;
import com.ndp.fb.mongo.api.EcReportMongoService;
import com.ndp.fb.rdb.api.ProductManageRdbService;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.velocity.app.VelocityEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.ui.velocity.VelocityEngineUtils;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * Created by dangw on 2016/4/25.
 */
@Service
public class EcReportMongoServiceImpl implements EcReportMongoService {
    private static Logger logger = LoggerFactory.getLogger(EcReportMongoServiceImpl.class);

    @Autowired
    private DB mongoDBClient;

    @Autowired
    private ProductManageRdbService productManageRdbService;

    @Autowired
    private VelocityEngine velocityEngine;

    // 页面传递的Filter字段和统计数据字段的对应关系
    private static Map<String, String> VIEW_FILTER_TO_STATS_FIELD_MAP = new HashMap() {
        {
            put("day", "day");
            put("webSite", "web_site");
            put("account", "accountId");
            put("adId", "adGroupId");
            put("adSetId", "campaignId");
            put("campaignId", "campaignGroupId");
            put("device", "convertedDevice");
            put("placement", "placement");
            put("age", "age");
            put("gender", "gender");
            put("country", "country");
            put("productId", "product_id");
        }
    };

    // 页面传递的Filter字段和业务数据字段的对应关系
    private static Map<String, String> VIEW_FILTER_TO_BUSINESS_FIELD_MAP = new HashMap() {
        {
            put("adType", "adType");
            put("adName", "name");
            put("adSetName", "adSet.name");
            put("campaignName", "campaign.name");
            put("optimizationGoal", "adSet.optimization_goal");
            put("billingEvent", "adSet.billing_event");
            put("autoBid", "adSet.is_autobid");
        }
    };

    // 页面传递的Filter字段中需要转换为数字类型的字段
    private static List<String> VIEW_FILTER_NUMBERS_FIELDS = new ArrayList() {
        {
            add("account");
            add("adId");
            add("adSetId");
            add("campaignId");
        }
    };

    // 页面传递的Filter字段中需要转换为Boolean类型的字段
    private static List<String> VIEW_FILTER_BOOL_FIELDS = new ArrayList() {
        {
            add("autoBid");
        }
    };

    // 页面传递的Group中需要变换为ID和名字的字段
    private static Map<String, String> VIEW_GROUP_TO_STATS_ID_FIELD_MAP = new HashMap() {
        {
            put("ad", "adGroupId");
            put("adSet", "campaignId");
            put("campaign", "campaignGroupId");
        }
    };

    private static Map<String, Object> STATS_FIELD_STATEMENT_FOR_PROJECT = new HashMap() {{
        put("spend", "$spend");
        put("impressions", "$impressions");
        put("clicks", "$clicks");
        put("cpc", new BasicDBObject("$cond", createConditionBasicDBObject(new BasicDBObject("$eq", new Object[]{"$clicks", 0}), "$spend", new BasicDBObject("$divide", new Object[]{"$spend", "$clicks"}))));
        put("cart", "$cart");
        put("checkout", "$checkout");
        put("checkoutCR", new BasicDBObject("$cond", createConditionBasicDBObject(new BasicDBObject("$eq", new Object[]{"$clicks", 0}), "$checkout", new BasicDBObject("$divide", new Object[]{"$checkout", "$clicks"}))));
        put("cps", new BasicDBObject("$cond", createConditionBasicDBObject(new BasicDBObject("$eq", new Object[]{"$checkout", 0}), "$spend", new BasicDBObject("$divide", new Object[]{"$spend", "$checkout"}))));
        put("salesAmount", "$salesAmount");
        put("roi", new BasicDBObject("$cond", createConditionBasicDBObject(new BasicDBObject("$eq", new Object[]{"$spend", 0}), "$salesAmount", new BasicDBObject("$divide", new Object[]{"$salesAmount", "$spend"}))));
    }};

    private static Map<String, String> STATS_FIELD_MAP_FOR_GROUP2 = new HashMap() {{
        put("impressions", "impressions");
        put("clicks", "clicks");
        put("spend", "spent");
        put("cart", "convertedArray.offsite_conversion.fb_pixel_add_to_cart.actions.1d_click");
        put("checkout", "convertedArray.offsite_conversion.fb_pixel_purchase.actions.1d_click");
        put("salesAmount", "convertedArray.offsite_conversion.fb_pixel_purchase.actionValues.1d_click");
    }};

    @Override
    public Map<String, Object> findFbDataFromMongo(String userTicket, EcReportInputParam ecReportInputParam) {
        List<String> groups = ecReportInputParam.getGroups();
        EcReportInputFilterParam filters = ecReportInputParam.getFilters();
        TimeRange statsTimeRange = filters.getDay();
        String statsTimeRangeSince = statsTimeRange.getSince();
        String statsTimeRangeUntil = statsTimeRange.getUntil();
        // 第一次match，主要根据统计数据所拥有的字段进行过滤，比如day,accountId,web_site,age,gender等
        BasicDBObject matchChild1 = new BasicDBObject()
                .append("day", new BasicDBObject().append("$gte", statsTimeRangeSince).append("$lte", statsTimeRangeUntil));    // 统计周期过滤
        String breakdowns = "all";  // 默认不使用细分数据
        if (groups.contains("device") || groups.contains("placement") || !CollectionUtils.isEmpty(filters.getDevice()) || !CollectionUtils.isEmpty(filters.getPlacement())) {
            breakdowns = "placement,impression_device";     //  设备和版位细分数据
        } else if (groups.contains("age") || groups.contains("gender") || !CollectionUtils.isEmpty(filters.getAge()) || !CollectionUtils.isEmpty(filters.getGender())) {
            breakdowns = "age,gender";  // 年龄，性别细分数据
        } else if (groups.contains("productId") || !CollectionUtils.isEmpty(filters.getProductId())) {
            breakdowns = "product_id";
        } else if (groups.contains("country") || !CollectionUtils.isEmpty(filters.getCountry())) {
            breakdowns = "country";     // 国家细分数据
        }
        matchChild1.put("breakdowns", breakdowns);

        if (CollectionUtils.isEmpty(ecReportInputParam.getFilters().getAccount())) {
            List<Long> fbAccountIds = productManageRdbService.findFbAccountIdByUserIdAndProductId(userTicket, ecReportInputParam.getFilters().getWebSite().get(0));
            matchChild1.put("accountId", new BasicDBObject("$in", fbAccountIds));
        }

        // 第一次group，根据adGroupId分组，减少后面Lookup业务数据的数量
        BasicDBObject groupChild1 = new BasicDBObject()
                .append("_id", "$adGroupId")
                .append("statsData", new BasicDBObject("$addToSet", "$$CURRENT"));
        // lookup，查询业务数据
        BasicDBObject lookupChild1 = new BasicDBObject()
                .append("from", "adgroup")
                .append("localField", "_id")
                .append("foreignField", "ad_id")
                .append("as", "adgroup");

        // 第一次project，提取业务数据中需要用到的字段
        BasicDBObject projectChild1 = new BasicDBObject()
                .append("ad_id", "$_id")
                .append("statsData", "$statsData")
                .append("adName", createLetBasicDBObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.name"));

        // 第二次match，主要根据业务数据，比如广告类型进行筛选
        BasicDBObject matchChild2 = new BasicDBObject();
        List<String> adType = filters.getAdType();
        if (groups.contains("adType") || !CollectionUtils.isEmpty(adType)) {
            projectChild1.put("adType", new BasicDBObject()
                    .append("$cond", createConditionBasicDBObject(new BasicDBObject("$eq", new Object[]{createLetBasicDBObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup.campaign.objective"), "PRODUCT_CATALOG_SALES"}), "dpa", "other")));
            if (!CollectionUtils.isEmpty(adType)) {
                matchChild2.put("adType", new BasicDBObject("$in", adType));
            }
        }

        // unwind，将统计数据拆分
        String unwindChild1 = "$statsData";

        // 第二次group，根据前端指定的分组条件进行分组统计数据（day,webSite,account,adId,adSetId,campaignId,device,placement,age,gender,country这些字段属于统计数据，adType,adName,adSetName,campaignName,optimizationGoal,billingEvent,autoBid属于业务字段）
        BasicDBObject idChildForGroup2 = new BasicDBObject();
        BasicDBObject groupChild2 = new BasicDBObject()
                .append("_id", idChildForGroup2)
                .append("adNames", new BasicDBObject("$addToSet", "$adName"))
                .append("adIds", new BasicDBObject("$addToSet", "$ad_id"));

        // 第二次project，计算cpc，cr等数据
        BasicDBObject projectChild2 = new BasicDBObject()
                .append("_id", 0)
                .append("day", "$_id.day")
                .append("adIds", "$adIds")
                .append("adNames", "$adNames");

        // 排序，暂时先按照day进行排序
        BasicDBObject sortChild1 = new BasicDBObject("day", 1);

        // group，第三次group，计算统计行的数据
        BasicDBObject groupChild3 = new BasicDBObject()
                .append("_id", null)
                .append("data", new BasicDBObject("$push", "$$CURRENT"))
                .append("count", new BasicDBObject("$sum", 1))
                .append("allAdNames", new BasicDBObject("$addToSet", "$adNames"))
                .append("allAdIds", new BasicDBObject("$addToSet", "$adIds"));

        // project，第三次project，分页
        Integer cPage = ecReportInputParam.getCPage();
        Integer pSize = ecReportInputParam.getPSize();
        int startNum = (cPage - 1) * pSize;
        BasicDBObject summaryChild = new BasicDBObject()
                .append("count", "$count");

        BasicDBObject projectChild3 = new BasicDBObject()
                .append("data", new BasicDBObject("$slice", new Object[]{"$data", startNum, pSize}))
                .append("summary", summaryChild)
                .append("allAdNames", "$allAdNames")
                .append("allAdIds", "$allAdIds")
                .append("allSpend", "$spend");

        // 以下是共通字段处理，需要修改时请特别注意
        JSONObject filtersJsonObject = JSON.parseObject(JSON.toJSONString(filters));    // 将过滤条件转为Map，以便统一处理

        final String finalBreakdowns = breakdowns;
        // 统计数据中的字段过滤以及名称变换
        VIEW_FILTER_TO_STATS_FIELD_MAP.forEach((key, value) -> {
            // 这里主要过滤数据
            if (!"day".equals(key) && !"webSite".equals(key)) {     // day和webSite已经提前处理好，跳过
                JSONArray filtersValue = filtersJsonObject.getJSONArray(key);
                if (!CollectionUtils.isEmpty(filtersValue)) {
                    if (VIEW_FILTER_NUMBERS_FIELDS.contains(key)) {
                        matchChild1.put(value, new BasicDBObject("$in", convertStrListToObjList(filtersValue, true, false)));
                    } else {
                        matchChild1.put(value, new BasicDBObject("$in", filtersValue));
                    }
                }
            }
            // 这里是为了返回与前端名字一致的字段名
            if (groups.contains(key)) {
                idChildForGroup2.put(key, "$statsData." + value);
                projectChild2.put(key, "$_id." + key);
            }

            // 这里对country和device,placement,age,gender的组合进行特殊处理（同时选择country,age等条件时，不直接使用country的细分数据，而使用age,gender等细分数据）
            if ("country".equals(key) && groups.contains("country") && !"country".equals(finalBreakdowns)) {
                JSONArray filtersValue = filtersJsonObject.getJSONArray(key);
                if (!CollectionUtils.isEmpty(filtersValue)) {
                    matchChild1.remove("country");   // 如果是这种特殊情况，那么国家就不在第一步进行过滤
                    matchChild2.put("country", new BasicDBObject("$in", filtersValue));   // 国家应该在取得业务数据上的国家之后再进行过滤
                }
                projectChild1.put("country",
                        new BasicDBObject()
                                .append("$let", new BasicDBObject()
                                        .append("vars", new BasicDBObject()
                                                .append("adgroup", new BasicDBObject()
                                                        .append("$arrayElemAt", new Object[]{"$adgroup", 0})))
                                        .append("in", new BasicDBObject()
                                                .append("$cond", new BasicDBObject()
                                                        .append("if", new BasicDBObject()
                                                                .append("$eq", new Object[]{new BasicDBObject().append("$arrayElemAt", new Object[]{"$$adgroup.adSet.targeting.geo_locations.countries", 0}), null}))
                                                        .append("then", new BasicDBObject()
                                                                .append("$arrayElemAt", new Object[]{"$$adgroup.adSet.targeting.geo_locations.regions.country", 0}))
                                                        .append("else", new BasicDBObject()
                                                                .append("$arrayElemAt", new Object[]{"$$adgroup.adSet.targeting.geo_locations.countries", 0}))))));
                idChildForGroup2.put("country", "$country");  // 由于在这种特殊情况下，country已经属于业务数据，因此需要覆盖其取值地方
            }
        });

        // 业务数据中的字段过滤以及名称变换
        VIEW_FILTER_TO_BUSINESS_FIELD_MAP.forEach((key, value) -> {
            // 这里主要过滤数据
            if (!"adType".equals(key)) {    // adType的判断比较特殊，已经提前处理好，跳过
                JSONArray filtersValue = filtersJsonObject.getJSONArray(key);
                if (groups.contains(key) || !CollectionUtils.isEmpty(filtersValue)) {
                    projectChild1.put(key, createLetBasicDBObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup." + value));
                    if (!CollectionUtils.isEmpty(filtersValue)) {
                        if (VIEW_FILTER_BOOL_FIELDS.contains(key)) {
                            matchChild2.put(key, new BasicDBObject("$in", convertStrListToObjList(filtersValue, false, true)));
                        } else {
                            matchChild2.put(key, new BasicDBObject("$in", filtersValue));
                        }
                    }
                }
            }
            // 这里是为了返回与前端名字一致的字段名
            if (groups.contains(key)) {
                idChildForGroup2.put(key, "$" + key);
                projectChild2.put(key, "$_id." + key);
            }
        });

        // Group分组字段中拼接业务字段
        VIEW_GROUP_TO_STATS_ID_FIELD_MAP.forEach((key, value) -> {
            if (groups.contains(key)) {
                projectChild1.put(key + "Name", createLetBasicDBObject(new BasicDBObject("adgroup", new BasicDBObject("$arrayElemAt", new Object[]{"$adgroup", 0})), "$$adgroup." + VIEW_FILTER_TO_BUSINESS_FIELD_MAP.get(key + "Name")));
                idChildForGroup2.put(key + "Id", "$statsData." + value);
                idChildForGroup2.put(key + "Name", "$" + key + "Name");
                projectChild2.put(key + "Id", "$_id." + key + "Id");
                projectChild2.put(key + "Name", "$_id." + key + "Name");
            }
        });

        // Group中拼接统计字段
        STATS_FIELD_MAP_FOR_GROUP2.forEach((key, value) -> {
            groupChild2.put(key, new BasicDBObject("$sum", "$statsData." + value));
            groupChild3.put(key, new BasicDBObject("$sum", "$" + key));
        });

        // Project中拼接统计字段
        STATS_FIELD_STATEMENT_FOR_PROJECT.forEach((key, value) -> {
            projectChild2.put(key, value);
            summaryChild.put(key, value);
        });

        // 组合步骤，总共11步
        List<DBObject> aggregateBasicDBObjects = new ArrayList<>();
        // 第1步，match，第一次match，主要根据统计数据所拥有的字段进行过滤，比如day,accountId,web_site,age,gender等
        aggregateBasicDBObjects.add(new BasicDBObject("$match", matchChild1));
        // 第2步，group，第一次group，根据adGroupId分组，减少后面Lookup业务数据的数量
        aggregateBasicDBObjects.add(new BasicDBObject("$group", groupChild1));
        // 第3步，lookup，查询业务数据
        aggregateBasicDBObjects.add(new BasicDBObject("$lookup", lookupChild1));
        // 第4步，project，第一次project，提取业务数据中需要用到的字段
        aggregateBasicDBObjects.add(new BasicDBObject("$project", projectChild1));
        // 第5步，unwind，将统计数据拆分
        aggregateBasicDBObjects.add(new BasicDBObject("$unwind", unwindChild1));
        // 第6步，match，第二次match，主要根据业务数据，比如广告类型进行筛选
        aggregateBasicDBObjects.add(new BasicDBObject("$match", matchChild2));
        // 第7步，group，根据前端指定的分组条件进行分组统计数据（day,webSite,account,adId,adSetId,campaignId,device,placement,age,gender,country这些字段属于统计数据，adType,adName,adSetName,campaignName,optimizationGoal,billingEvent,autoBid属于业务字段）
        aggregateBasicDBObjects.add(new BasicDBObject("$group", groupChild2));
        // 第8步，project，第二次project，计算cpc，cr等数据
        aggregateBasicDBObjects.add(new BasicDBObject("$project", projectChild2));
        // 第9步，sort，暂时先按照day进行排序
        aggregateBasicDBObjects.add(new BasicDBObject("$sort", sortChild1));
        // 第10步，group，第三次group，计算统计行的数据
        aggregateBasicDBObjects.add(new BasicDBObject("$group", groupChild3));
        // 第11步，project，第三次project，分页
        aggregateBasicDBObjects.add(new BasicDBObject("$project", projectChild3));
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateBasicDBObjects));

        long start = System.currentTimeMillis();
        Map result = null;
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateBasicDBObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        try {
            if (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                if (null != dbObject) {
                    result = dbObject.toMap();
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    @Override
    public Map<String, Object> findFbDataFromMongoForABTest(EcReportForABTestInputParam ecReportForABTestInputParam) {
        Map<String, Object> model = new HashMap<String, Object>();
        // 页数转换
        ecReportForABTestInputParam.setCPage((ecReportForABTestInputParam.getCPage() - 1) * ecReportForABTestInputParam.getPSize());
        model.putAll(JSON.parseObject(JSON.toJSONString(ecReportForABTestInputParam)));
        String querySentence = VelocityEngineUtils.mergeTemplateIntoString(velocityEngine, "EcReportForABTest.vm", "UTF-8", model).replace("\\", "");
        logger.info("Velocity替换变量后语句:" + querySentence);
        BasicDBList queryList = (BasicDBList) com.mongodb.util.JSON.parse(querySentence);
        List<DBObject> aggregateBasicDBObjects = new ArrayList<>();
        queryList.forEach(query -> {
            aggregateBasicDBObjects.add((BasicDBObject) query);
        });
        logger.info("Mongo查询语句:" + com.mongodb.util.JSON.serialize(aggregateBasicDBObjects));

        long start = System.currentTimeMillis();
        Map result = new HashMap<>();
        AggregationOptions aggregationOptions = AggregationOptions.builder().allowDiskUse(true).build();
        Cursor cursor = mongoDBClient.getCollection(MongoConstants.TABLE_NAME_ADGROUP_STATS_BY_DAY).aggregate(aggregateBasicDBObjects, aggregationOptions);
        logger.info("Mongo聚合查询花费时间：" + (System.currentTimeMillis() - start) + "毫秒");
        try {
            if (cursor.hasNext()) {
                DBObject dbObject = cursor.next();
                if (null != dbObject) {
                    result = dbObject.toMap();
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            cursor.close();
        }
        return result;
    }

    /**
     * 将String数组转换为Object数组，其中数字会被强转为Long型，true/false强转为Boolean型，否则还是String
     *
     * @param srcList     原始数组
     * @param convertNum  是否要转换数字
     * @param convertBool 是否要转换Bool
     * @return
     */
    private List<Object> convertStrListToObjList(JSONArray srcList, boolean convertNum, boolean convertBool) {
        List<Object> targetList = new ArrayList<>();
        srcList.forEach(src -> {
            String str = src.toString();
            if (NumberUtils.isNumber(str)) {
                if (convertNum) {
                    targetList.add(NumberUtils.createLong(str));
                } else {
                    targetList.add(str);
                }
            } else if (null != BooleanUtils.toBooleanObject(str)) {
                if (convertBool) {
                    targetList.add(BooleanUtils.toBooleanObject(str));
                } else {
                    targetList.add(str);
                }
            } else {
                targetList.add(str);
            }
        });
        return targetList;
    }

    private BasicDBObject createLetBasicDBObject(Object varsObject, Object inObject) {
        return new BasicDBObject("$let", new BasicDBObject().append("vars", varsObject).append("in", inObject));
    }

    private static BasicDBObject createConditionBasicDBObject(Object ifObject, Object thenObject, Object elseObject) {
        BasicDBObject condDBObject = new BasicDBObject();
        condDBObject.put("if", ifObject);
        condDBObject.put("then", thenObject);
        condDBObject.put("else", elseObject);
        return condDBObject;
    }
}
